package com.itzhoun.client;

import com.itzhoun.callback.MyMQTTCallback;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author zhounan
 * @date 2025/01/23 16:10
 * @Description
 */
public class MyMQTTClient {

    private static final Logger LOGGER = LoggerFactory.getLogger(MyMQTTClient.class);

    private static MqttClient client;
    private String host;
    private String username;
    private String password;
    private String clientId;
    private int timeout;
    private int keepalive;

    public MyMQTTClient(String host, String username, String password, String clientId, int timeout, int keepalive) {
        this.host = host;
        this.username = username;
        this.password = password;
        this.clientId = clientId;
        this.timeout = timeout;
        this.keepalive = keepalive;
    }

    public static MqttClient getClient() {
        return client;
    }


    public static void setClient(MqttClient client){
        MyMQTTClient.client = client;
    }


    /**
     *
     * 设置mqtt连接参数
     * @param username 用户铭
     * @param password 密码
     * @param timeout 超时时间
     * @param keepalive 存活时间
     */
    public MqttConnectOptions setMqttConnectOptions(String username, String password, int timeout, int keepalive) {
        MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
        mqttConnectOptions.setUserName(username);
        mqttConnectOptions.setPassword(password.toCharArray());
        mqttConnectOptions.setConnectionTimeout(timeout);
        mqttConnectOptions.setKeepAliveInterval(keepalive);

        //表示开启会话清除模式 意味着会话信息不会被保留 每次连接都是作为一个新会话的开始、无持久化信息
        mqttConnectOptions.setCleanSession(true);
        //表示开启自动重连机制 意味着连接中断检测、尝试重新连接、恢复会话状态
        mqttConnectOptions.setAutomaticReconnect(true);
        return mqttConnectOptions;
    }


    /**
     *
     * 连接mqtt服务端，得到MqttClient对象
     */
    public void connect() throws MqttException {
        if (client == null){
            //MemoryPersistence 核心目的是确保在客户端意外断开连接或重启时、未发送或未确认的消息不会丢失
            client = new MqttClient(host, clientId, new MemoryPersistence());

            client.setCallback(new MyMQTTCallback(this));
        }
        MqttConnectOptions mqttConnectOptions = setMqttConnectOptions(username, password, timeout, keepalive);
        if (!client.isConnected()){
            client.connect(mqttConnectOptions);
        }else {
            client.disconnect();
            client.connect(mqttConnectOptions);
        }
        //未发生异常，则连接成功
        LOGGER.info("MQTT connect success");

    }


    //发布消息
    public void publish(String pushMessage, String topic, int qos, boolean retained){
        MqttMessage mqttMessage = new MqttMessage();
        mqttMessage.setPayload(pushMessage.getBytes());
        mqttMessage.setQos(qos);
        mqttMessage.setRetained(retained);
        MqttTopic mqttTopic = MyMQTTClient.getClient().getTopic(topic);
        if (mqttTopic == null){
            LOGGER.error("topic is not exist");
        }

        //配送
        MqttDeliveryToken deliveryToken;
        synchronized (this){
            try {
                deliveryToken = mqttTopic.publish(mqttMessage);
                deliveryToken.waitForCompletion(1000L);
            } catch (MqttPersistenceException e) {
                e.printStackTrace();
                throw new RuntimeException(e);
            } catch (MqttException e) {
                e.printStackTrace();
                throw new RuntimeException(e);
            }
        }
    }

    /**
     *
     * 订阅某个主题
     * @param topic 主题
     * @param qos
     */
    public void subscribe(String topic, int qos){
        try {
            MyMQTTClient.getClient().subscribe(topic, qos);
            LOGGER.info("订阅主题:{}成功", topic);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }


    /**
     *
     * 取消某个主题
     */
    public void cleanTopic(String topic){
        if (client == null || !client.isConnected()){
            throw new RuntimeException("客户端未连接成功");
        }

        try {
            client.unsubscribe(topic);
            LOGGER.info("取消订阅主题:{}成功", topic);
        } catch (MqttException e) {
            e.printStackTrace();
        }

    }

    public void publish(String msg, String topic) {
        this.publish(msg, topic, 0, false);
    }
}
